package com.jianying.day07;

import com.jianying.beans.WaterSensor;
import com.jianying.func.WaterSensorMapFunction;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.streaming.api.TimerService;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.util.Collector;

/**
 * ClassName: Flink04_Timer
 * Package: com.atguigu.day07
 * Description:
 *
 * @Author: tubage
 * @Create: 2024/4/8 11:38
 * @Version: 1.0
 */
public class Flink04_Timer {
    public static void main(String[] args) throws Exception {
        // TODO 定时器
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        SingleOutputStreamOperator<WaterSensor> wsDS = env.socketTextStream("localhost", 8888)
                .map(new WaterSensorMapFunction())
                .assignTimestampsAndWatermarks(
                        WatermarkStrategy
                                .<WaterSensor>forMonotonousTimestamps()
                                .withTimestampAssigner(
                                        new SerializableTimestampAssigner<WaterSensor>() {
                                            @Override
                                            public long extractTimestamp(WaterSensor waterSensor, long l) {
                                                return waterSensor.getTs();
                                            }
                                        }
                                )

                );
        
        // TODO 必须先keyBy才能使用
        KeyedStream<WaterSensor, String> keyDS = wsDS.keyBy(WaterSensor::getId);
        
        keyDS.process(
                new KeyedProcessFunction<String, WaterSensor, String>() {
                    @Override
                    public void processElement(WaterSensor value, Context ctx, Collector<String> out) throws Exception {
                        // 处理函数继承了富函数类 有上下文、生命周期方法
                        Long timestamp = ctx.timestamp(); // 当前元素对应的事件时间，
                        System.out.println("timestamp = " + timestamp);
                        String currentKey = ctx.getCurrentKey(); // 获取当前分组的key
                        System.out.println("currentKey = " + currentKey);
                        // ctx.output(); 侧输出流必须使用process
                        // ---------------------------
                        TimerService timerService = ctx.timerService(); // 获取时间服务
                        long watermark = timerService.currentWatermark();
                        System.out.println("watermark = " + watermark); // 水位线

                        long currentProcessingTime = timerService.currentProcessingTime(); // 处理时间
                        // TODO 注册处理时间定时器
                        timerService.registerProcessingTimeTimer(currentProcessingTime + 10000);
                        timerService.deleteProcessingTimeTimer(currentProcessingTime + 10000);

                        // TODO 注册定时器 事件时间
                        timerService.registerEventTimeTimer(10);
                        if(true) {
                            timerService.deleteEventTimeTimer(10); // 根据条件删除定时器
                        }
                    }

                    // TODO  定时器触发时调用
                    @Override
                    public void onTimer(long timestamp, KeyedProcessFunction<String, WaterSensor, String>.OnTimerContext ctx, Collector<String> out) throws Exception {
                        String currentKey = ctx.getCurrentKey();
                        System.out.println("key=" + currentKey + "现在时间是" + timestamp + "定时器触发");
                    }
                }
        ).print();

        env.execute();

    }
}
